// 代理类封装了MqttSubConnection作为其委托类，实现所有上层逻辑
#include "MqttConnection.h"
void inline MqttConnection::tryEmplace(std::string topic, MqttOnMessageCallback *callback)
{
    if (!topicCallbackMap.count(topic))
    {
        topicCallbackMap.emplace(topic, std::unordered_set<MqttOnMessageCallback *>());
    }
    topicCallbackMap.at(topic).emplace(callback);
}
bool inline MqttConnection::tryRemove(std::string topic, MqttOnMessageCallback *callback)
{
    if (!topicCallbackMap.count(topic))
    {
        return true;
    }
    topicCallbackMap.at(topic).erase(callback);
    if (topicCallbackMap.at(topic).empty())
    {
        topicCallbackMap.erase(topic);
        return true;
    }
    return false;
}
bool MqttConnection::connect()
{
    if (!enable)
        return false;
    // ifprintf("lk before connect\n");
    std::lock_guard<std::recursive_mutex> lk(subMutex);
    if (connection != nullptr)
    {
        delete connection;
    }
    this->clientId = std::string(MQTT_PREFIX).append(userName).append("_").append(this->uuid.substr(0, 4));
    this->connection = new MqttSubConnection(&topicCallbackMap, &connected, clientId, true);
    if (address.empty())
    {
        return false;
    }
    if (port <= 0 || port >= 65535)
    {
        return false;
    }

    if (!address.empty())
    {
        if (!password.empty())
        {
            int res = connection->username_pw_set(userName.c_str(), password.c_str());
            ifprintf("set[%s]result:(%d)%s\n", userName.c_str(), res, mosqpp::strerror(res));
        }
        else
        {
            int res = connection->username_pw_set(userName.c_str());
            ifprintf("set[%s]result:(%d)%s\n", userName.c_str(), res, mosqpp::strerror(res));
        }
    }
    int res = connection->connect(address.c_str(), port);
    ifprintf("connect[%s]result:(%d)%s\n", this->name.c_str(), res, mosqpp::strerror(res));
    usleep(1000 * 500);
    if (res == 0)
    {
        connected = true;
        connection->subscribeAll();
    }
    else
    {
        connected = false;
    }
    // connection->subscribeAll();
    this->looping = false;
    return true;
}

inline long MqttConnection::getCurrentMs()
{
    struct timeval tv;
    struct timezone tz;
    gettimeofday(&tv, &tz);
    return tv.tv_sec * 1000 + tv.tv_usec / 1000;
}
MqttConnection::~MqttConnection()
{
    this->looping = true;
    std::lock_guard<std::recursive_mutex> lk(subMutex);
    delete connection;
}
MqttConnection::MqttConnection(std::unordered_map<std::string, MqttConnection *> *nameConnectionMap,
                               const std::string &name,
                               const std::string &address,
                               bool enable,
                               int port,
                               const std::string &userName,
                               const std::string &password)
{
    this->nameConnectionMap = nameConnectionMap;
    connection = nullptr;
    this->name = name;
    this->address = address;
    this->port = port;
    this->userName = userName;
    this->password = password;
    this->looping = false;
    this->clientId = "";
    this->connected = false;
    this->enable = enable;
    nameConnectionMap->emplace(name, this);
}
std::string MqttConnection::getClientId()
{
    ifprintf("%s\n", this->clientId.c_str());
    return this->clientId;
}
bool MqttConnection::subscribe(const std::string &topic, MqttOnMessageCallback *callback)
{
    tryEmplace(topic, callback);
    // ifprintf("lk before subscribe\n");
    // std::lock_guard<std::recursive_mutex> lk(subMutex);
    // ifprintf("lk after subscribe\n");
    int res = connection->subscribe(NULL, topic.c_str());
    ifprintf("subscribe[%s]result:(%d)%s\n", this->name.c_str(), res, mosqpp::strerror(res));
    if (res)
        return false;
    return true;
}
void MqttConnection::unsubscribe(std::string topic, MqttOnMessageCallback *callback)
{
    if (tryRemove(topic, callback))
    {
        // ifprintf("lk before unsubscribe\n");
        // std::lock_guard<std::recursive_mutex> lk(subMutex);
        // ifprintf("lk after unsubscribe\n");
        connection->unsubscribe(NULL, topic.c_str());
    }
}
bool MqttConnection::publish(const std::string &topic, const std::string &data)
{
    // ifprintf("lk before publish\n");
    // std::lock_guard<std::recursive_mutex> lk(subMutex);
    // ifprintf("lk after publish\n");
    if (!enable)
        return false;
    if (topic == "")
    {
        return false;
    }
    int res = connection->publish(NULL, topic.c_str(), data.size(), data.c_str());
    ifprintf("publish[%s]result:(%d)%s\n", this->name.c_str(), res, mosqpp::strerror(res));
    if (res)
        return false;
    return true;
}
void MqttConnection::loop(MqttConnection *p)
{
    if (!p->enable)
        return;
    if (p->looping)
        return;
    std::lock_guard<std::recursive_mutex> lk(p->subMutex);
    p->looping = true;
    if (p->connection == nullptr || p->connected == false)
    {
        p->connect();
        return;
    }
    int res = p->connection->loop(5);
    if (res != 0)
    {
        p->connected = false;
    }
    ifprintf("loop[%s]result:(%d)%s\n", p->name.c_str(), res, mosqpp::strerror(res));
    p->looping = false;
}
void MqttConnection::setAddress(const std::string &address, bool autoConnect)
{
    if (address == this->address)
    {
        return;
    }
    else
    {
        this->address = address;
        if (autoConnect)
        {
            connect();
        }
    }
}
void MqttConnection::setPort(int port, bool autoConnect)
{
    if (port == this->port)
    {
        return;
    }
    else
    {
        this->port = port;
        if (autoConnect)
        {
            connect();
        }
    }
}
void MqttConnection::setUserName(const std::string &userName, bool autoConnect)
{
    if (userName == this->userName)
    {
        return;
    }
    else
    {
        this->userName = userName;
        if (autoConnect)
        {
            connect();
        }
    }
}
void MqttConnection::setEnable(bool enable)
{
    if (this->enable == enable)
    {
        return;
    }
    this->enable = enable;
    if (!this->enable)
    {
        this->connection->disconnect();
    }
    else
    {
        this->connection->connect(address.c_str(), port);
    }
}
void MqttConnection::setPassword(const std::string &password, bool autoConnect)
{
    if (password == this->password)
    {
        return;
    }
    else
    {
        this->password = password;
        if (autoConnect)
        {
            connect();
        }
    }
}
bool MqttConnection::setName(const std::string &name)
{
    if(this->name==name) return true;
    if (nameConnectionMap != nullptr)
    {
        if (nameConnectionMap->count(name))
            return false;
        nameConnectionMap->erase(this->name);
        nameConnectionMap->emplace(name, this);
    }
    this->name = name;
    return true;
}
std::string MqttConnection::getName()
{
    return this->name;
}
std::string MqttConnection::getStatus()
{
    if (!this->enable)
        return "DISABLE";
    if (!this->connected)
        return "BAD";
    return "GOOD";
}
std::string MqttConnection::getAddress()
{
    return this->address;
}
int MqttConnection::getPort()
{
    return this->port;
}
bool MqttConnection::isEnable()
{
    return this->enable;
}
std::string MqttConnection::getUserName()
{
    return this->userName;
}
std::string MqttConnection::getPassword()
{
    return this->password;
}